package com.concurrency.cf.util;

import cn.hutool.core.collection.CollectionUtil;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
public class CollUtil {

    public static void main(String[] args) {

        ExecutorService executorService = Executors.newCachedThreadPool();

        List<Integer> ids = Lists.newArrayList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

        List<List<Integer>> lists = partitionProcess(ids, 3, integers -> integers, true, executorService);

        log.info(lists.toString());

        List<List<Integer>> list1 = new ArrayList<>();
        try {
            list1 = partitionProcess(ids, 3, integers -> {
                throw new RuntimeException("11111");
//                return integers;
            }, true, executorService);
        } catch (Exception e) {
            executorService.shutdown();
            log.error(e.getMessage());
            return;
        }

        log.info(list1.toString());

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        executorService.shutdown();
    }

    /**
     * 分批执行方法
     *
     * @param ids      id列表
     * @param size     分批的数量
     * @param func     需要执行的方法
     * @param fastFail 是否快速失败，true-异常直接抛出，false-忽略其中的异常
     * @param executor 线程池
     * @param <R>      出参类型
     * @param <T>      入参类型
     * @return
     */
    public static <R, T> List<R> partitionProcess(List<T> ids, int size, Function<List<T>, R> func, boolean fastFail, Executor executor) {
        if (CollectionUtil.isEmpty(ids)) {
            return Collections.EMPTY_LIST;
        }

        List<T> distinctList = ids.stream().distinct().collect(Collectors.toList());
        List<List<T>> partition = Lists.partition(distinctList, size);
        List<Throwable> list = new ArrayList<>();
        CompletableFuture[] cfs = partition.stream()
                .map(idList -> doSupplyAsync(func, executor, list, idList))
                .filter(Objects::nonNull)
                .toArray(CompletableFuture[]::new);
        if (fastFail && list.size() > 0) {
            throw new RuntimeException(list.get(0).getMessage());
        }
        return Arrays.stream(cfs)
                .map(cf -> (R) doGet(cf))
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
    }

    private static <R, T> CompletableFuture<R> doSupplyAsync(Function<List<T>, R> func, Executor executor, List<Throwable> list, List<T> idList) {
        log.info("start supplyAsync");
        CompletableFuture<R> exceptionally = CompletableFuture.supplyAsync(() -> doApply(func, idList), executor)
                .exceptionally(e -> {
                    list.add(e);
                    return null;
                });
        log.info("end supplyAsync, exceptionally:{}", exceptionally);
        return exceptionally;
    }

    private static <R, T> R doApply(Function<List<T>, R> func, List<T> idList) {
        log.info("start apply");
        R result = func.apply(idList);
        log.info("end apply, result:{}", result);
        return result;
    }

    private static Object doGet(CompletableFuture cf) {
        try {
            return cf.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        return null;
    }

}
